package com.dj.ms.event;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.dj.ms.configure.EventBusConfiguration;
import com.dj.ms.event.support.*;
import com.dj.ms.support.SpringBeanFactoryUtils;
import com.google.common.collect.Queues;
import com.google.common.eventbus.AsyncEventBus;
import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.SubscriberExceptionContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.lang.reflect.Method;
import java.util.Date;
import java.util.Objects;
import java.util.Queue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * <br>
 * <p>事件处理handle</p>
 *
 * <br>
 *
 * @author ZJG
 * @version 1.0
 * @date 2021/7/26 下午6:04
 */
@Slf4j
@Component
@ConditionalOnExpression("${dj.common.event.enabled:false}")
@ConditionalOnClass({EventBusConfiguration.class, EventRetryTaskProperties.class})
@EnableConfigurationProperties(EventRetryTaskProperties.class)
public class EventBusHandler {

    @Autowired
    private EventRetryTaskProperties eventRetryTaskProperties;

    @Autowired
    private EventBusConfiguration eventBusConfiguration;

    private static EventBusHandler eventBusHandler;

    @PostConstruct
    public void init() {
        eventBusHandler = this;
        eventBusHandler.eventRetryTaskProperties = this.eventRetryTaskProperties;
        eventBusHandler.eventBusConfiguration = this.eventBusConfiguration;
    }

    /**
     * 同步事件
     */
    private final static EventBus syncEventBus = new EventBus("syncEventBus");

    /**
     * 异步事件
     */
    private final static AsyncEventBus asyncEventBus = new AsyncEventBus(
            new AsyncEventBusExecutor(getProperties().getCorePs().intValue(), getProperties().getMaxPs().intValue()),
            (exception, context) -> {
                if (Objects.nonNull(true)) {
                    saveTask(context, exception);
                }
            });

    /**
     * 长耗时任务事件总线
     */
    private final static AsyncEventBus taskEventBus = new AsyncEventBus(
            new AsyncEventBusExecutor(getProperties().getCorePs().intValue(),
                    getProperties().getMaxPs().intValue(),
                    getProperties().getQueueSize().intValue(),
                    "taskEventBus",
                    true),
            (exception, context) -> {
                if(Objects.nonNull(true)){
                    saveTask(context, exception);
                }
            });

    /**
     * 异步事件队列
     */
    private static final ThreadLocal<Queue<Object>> queue = ThreadLocal.withInitial(() -> Queues.newArrayDeque());

    /**
     * 是否在事务中
     */
    public static final ThreadLocal<Boolean> inTransaction = ThreadLocal.withInitial(() -> false);

    /**
     * 定时线程池
     */
    private static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = new ScheduledThreadPoolExecutor(
            2,
            new BasicThreadFactory
                    .Builder()
                    .namingPattern("e-retry-pool-%d")
                    .daemon(true)
                    .build());

    private static boolean saveTask(SubscriberExceptionContext context, Throwable exception) {
        EventRetryTask eventRetryTask = null;
        if (Objects.nonNull(EventBusHandler.getConfig())
                && EventBusHandler.getConfig().isRetryEnabled()
                && Objects.nonNull(context)) {
            String canonicalName = context.getSubscriber().getClass().getCanonicalName();
            if (Objects.isNull(EventRetryTaskDataManager.getRepository(canonicalName))) {
                return false;
            }
            eventRetryTask = new EventRetryTask();
            eventRetryTask.setCreateTime(new Date());
            eventRetryTask.setClassName(canonicalName);
            Method method = context.getSubscriberMethod();
            eventRetryTask.setMethod(method.getName());
            eventRetryTask.setParameterClassName(method.getParameterTypes()[0].getCanonicalName());
            eventRetryTask.setEvent(context.getEvent());
            eventRetryTask.setId(String.valueOf(System.currentTimeMillis()));
            eventRetryTask.setExceptionMessage(exception.getMessage());
            eventRetryTask.setExceptionStackTrace(ExceptionUtils.getStackTrace(exception));
            EventRetryTaskDataManager.getRepository(canonicalName).save(eventRetryTask);
            log.error("Exception thrown by subscriber method {}'('{}')' on subscriber {} when dispatching event: {}",
                    method.getName(),
                    method.getParameterTypes()[0].getName(),
                    context.getSubscriber(),
                    context.getEvent(),
                    exception);
            return true;
        }
        log.warn("EventBusHandler::未找到EventRetryTaskRepository实例. eventRetryTask={}",
                JSONObject.toJSONString(eventRetryTask));
        return false;
    }

    /**
     * 提交一个没有事务的事件,先提交同步事件，再提交异步事件
     *
     * @param event 事件对象
     */
    public static void postNoTransaction(Object event) {
        if (Objects.isNull(event)) {
            return;
        }
        String eventClassName = Objects.isNull(event) ? "null" : event.getClass().getSimpleName();
        postSyncEvent(event, eventClassName);
        asyncEventBus.post(event);
        taskEventBus.post(event);
        if(log.isInfoEnabled()) {
            log.info("异步事件已发布，{}:{}", eventClassName, JSON.toJSON(event));
        }

    }

    /**
     * 提交一个事件,先提交同步事件，再提交异步事件
     *
     * @param event 事件对象
     */
    public static void post(Object event) {
        String eventClassName = Objects.isNull(event) ? "null" : event.getClass().getSimpleName();
        postSyncEvent(event, eventClassName);

        Queue<Object> queueForThread = queue.get();
        queueForThread.offer(event);
        if(log.isDebugEnabled()) {
            log.debug((!inTransaction.get() ? "未" : "已") + "在事务中，预存异步事件，{}:{}", eventClassName, JSON.toJSON(event));
        }

        if(!inTransaction.get()) {
            SCHEDULED_EXECUTOR_SERVICE.schedule(() -> {
                if(queueForThread.contains(event)) {
                    asyncEventBus.post(event);
                    taskEventBus.post(event);
                    queueForThread.remove(event);
                    log.warn("没有事务，定时补发异步事件，{}:{}", event.getClass().getSimpleName(), JSON.toJSON(event));
                }
            }, 3, TimeUnit.SECONDS);
        }
    }

    /**
     * 注册同步监听器
     */
    public static void registerSync(Object object) {
        syncEventBus.register(object);
    }
    /**
     * 注册异步监听器
     */
    public static void registerAsync(Object object) {
        asyncEventBus.register(object);
    }
    /**
     * 注册超长时间执行任务的监听器
     */
    public static void registerTask(Object object) {
        taskEventBus.register(object);
    }

    public static EventRetryTaskProperties getProperties() throws BeansException {
        EventRetryTaskProperties eventRetryTaskProperties = SpringBeanFactoryUtils.getBean(EventRetryTaskProperties.class);
        if (Objects.nonNull(eventRetryTaskProperties)) {
            return eventRetryTaskProperties;
        }
        return null;
    }

    public static EventBusConfiguration getConfig() throws BeansException {
        EventBusConfiguration eventBusConfiguration = SpringBeanFactoryUtils.getBean(EventBusConfiguration.class);
        if (Objects.nonNull(eventBusConfiguration)) {
            return eventBusConfiguration;
        }
        return null;
    }

    /**
     * 发布同步事件
     * @param event
     * @param eventClassName
     */
    private static void postSyncEvent(Object event, String eventClassName) {
        try {
            if (log.isInfoEnabled()) {
                log.info("同步事件开始，{}:{}", eventClassName, JSON.toJSON(event));
            }
            syncEventBus.post(event);
            if (log.isInfoEnabled()) {
                log.info("同步事件结束，{}:{}", eventClassName, JSON.toJSON(event));
            }
        } catch (Error e) {
            log.error("同步事件异常，{}:{}", eventClassName, JSON.toJSON(event), e);
            throw e;
        }
    }

    /**
     * 触发队列中的异步事件
     */
    public static void doAsyncPost(){
        Queue<Object> queueForThread = queue.get();
        if(queueForThread == null || queueForThread.isEmpty()){
            return;
        }
        Object nextEvent = null;
        try {
            while ((nextEvent = queueForThread.poll()) != null) {
                asyncEventBus.post(nextEvent);
                taskEventBus.post(nextEvent);
                if(log.isInfoEnabled()) {
                    log.info("事务完成，已触发异步事件，{}:{}", nextEvent.getClass().getSimpleName(), JSON.toJSON(nextEvent));
                }
            }
        }catch (Exception e){
            log.error("使用事务，异步事件触发异常，{}:{}", nextEvent != null ? nextEvent.getClass().getSimpleName() : "null", JSON.toJSON(nextEvent), e);
        }finally {
            queue.remove();
        }
    }

}
